package comparer

import (
	"log"
	"sync"
	"sync/atomic"

	"mig/pkg/chunk"
	"mig/pkg/data"
	"mig/pkg/types"
	"mig/pkg/utils"
)

func (c *CommonComparer) commonCompare(src data.Database, tgt data.Database) (*types.CompareResult, error) {
	srcTotalRows, err := src.Count()
	if err != nil {
		return nil, err
	}
	tgtTotalRows, err := tgt.Count()
	if err != nil {
		return nil, err
	}
	totalRows := max(srcTotalRows, tgtTotalRows)
	log.Printf("total rows: %d", totalRows)
	totalChunks := totalRows / c.chunkSize
	if totalRows%c.chunkSize > 0 {
		totalChunks++
	}

	chunkNo := int64(-1)
	wg := sync.WaitGroup{}
	var unequalChunks []types.UnequalChunk
	mtx := sync.Mutex{}
	for i := 0; i < c.maxWorkers; i++ {
		wg.Add(1)
		go func() {
			var (
				srcChunk  chunk.Chunk
				tgtChunk  chunk.Chunk
				workerErr error
			)

			for {
				currentNo := atomic.AddInt64(&chunkNo, 1)
				if currentNo >= totalChunks {
					wg.Done()
					return
				}

				srcChunk, workerErr = src.ReadChunk(currentNo, c.chunkSize)
				if workerErr != nil {
					log.Printf("read error: %v", workerErr)
					continue
				}

				tgtChunk, workerErr = tgt.ReadChunk(currentNo, c.chunkSize)
				if workerErr != nil {
					log.Printf("read error: %v", workerErr)
					continue
				}

				equal := true
				for idx, record := range srcChunk.Records() {
					if !utils.EqualSlice(record.Bytes(), tgtChunk.Records()[idx].Bytes()) {
						equal = false
						break
					}
				}

				if !equal {
					mtx.Lock()
					unequalChunks = append(unequalChunks, types.UnequalChunk{
						ChunkNo: int(currentNo),
						Node:    tgtChunk.Digest(),
					})
					mtx.Unlock()
				}
			}
		}()
	}
	wg.Wait()
	result := &types.CompareResult{IsComparable: true, UnequalChunks: unequalChunks}
	return result, nil
}

type CommonComparer struct {
	chunkSize  int64
	maxWorkers int
}

func NewCommonComparer(chunkSize int64, maxWorkers int) *CommonComparer {
	return &CommonComparer{
		chunkSize:  chunkSize,
		maxWorkers: maxWorkers,
	}
}

func (c *CommonComparer) Compare(src data.Database, tgt data.Database,
	srcInfo *types.IntegrityInfo) (*types.CompareResult, error) {
	return c.commonCompare(src, tgt)
}
